#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <fcntl.h>    // For O_* constants
#include <sys/stat.h> // For mode constants
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <sys/time.h>
#include <sys/mman.h>
#include <semaphore.h>
#include <string.h>

#define SEM_READ_NAME(name) "/my_rsem_" #name
#define SEM_WRITE_NAME(name) "/my_wsem_" #name
#define INIT_FLAG 0xA5A5A5A5
#define ITEM_HDR 2

// 队列放在共享内存中，收发双方都可以操作
typedef struct
{
    // 队列初始化标志
    unsigned int init_flag;
    // 当前队列可读元素序号
    unsigned int pos_r;
    // 当前队列可写元素序号
    unsigned int pos_w;
    // 队列中元素最大存储大小
    unsigned int size;
    // 队列中元素个数
    unsigned int num;
} queue_comm_t;

// 放在每个单独进程中
typedef struct
{
    char *name;
    // 资源可读信号量，用于同步
    sem_t *sem_can_read;
    // 资源可写信号量，用于同步
    sem_t *sem_can_write;
    // 共享内存
    unsigned char *data;
    // 队列公共数据，放在共享内存头部(内部不能包含指针)
    queue_comm_t *comm;
} queue_t;

static int delayms(int ms)
{
    int ret;
    struct timeval timeout;
    timeout.tv_sec = ms / 1000;
    timeout.tv_usec = (ms % 1000) * 1000;
    do
    {
        ret = select(0, NULL, NULL, NULL, &timeout);
        if (ret < 0 && errno == EINTR)
        {
            // printf("signal break,continue\n");
            continue;
        }
        else if (ret < 0)
        {
            return -1;
        }
    } while (ret != 0);
    return 0;
}

void *process_queue_create(char *name, int size, int num)
{
    char *shm_ptr = NULL;
    sem_t *sem_can_read = NULL;
    sem_t *sem_can_write = NULL;
    queue_t *p_queue = NULL;
    queue_comm_t *p_queue_comm = NULL;
    int shm_fd = -1, total_size = 0;

    p_queue = malloc(sizeof(queue_t));
    if (p_queue == NULL)
    {
        return NULL;
    }
    shm_fd = shm_open(name, O_CREAT | O_RDWR, 0666);
    if (shm_fd == -1)
    {
        perror("shm_open");
        goto failed;
    }

    //  配置共享内存大小
    total_size = (size + ITEM_HDR) * num + sizeof(queue_t);
    if (ftruncate(shm_fd, total_size) == -1)
    {
        perror("ftruncate");
        goto failed;
    }
    // 将共享内存映射到进程地址空间
    shm_ptr = mmap(0, total_size, PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
    if (shm_ptr == MAP_FAILED)
    {
        perror("mmap");
        goto failed;
    }
    close(shm_fd);

    // 打开命名信号量
    sem_can_read = sem_open(SEM_READ_NAME(name), O_CREAT, 0644, 0);
    if (sem_can_read == SEM_FAILED)
    {
        perror("sem_open1");
        goto failed;
    }

    sem_can_write = sem_open(SEM_WRITE_NAME(name), O_CREAT, 0644, num);
    if (sem_can_write == SEM_FAILED)
    {
        perror("sem_open2");
        goto failed;
    }

    p_queue_comm = (queue_comm_t *)shm_ptr;
    p_queue->name = name;
    p_queue->sem_can_read = sem_can_read;
    p_queue->sem_can_write = sem_can_write;
    p_queue->data = shm_ptr + sizeof(queue_comm_t);
    p_queue->comm = p_queue_comm;

    if (p_queue_comm->init_flag != INIT_FLAG)
    {
        p_queue_comm->pos_r = 0;
        p_queue_comm->pos_w = 0;
        p_queue_comm->init_flag = INIT_FLAG;
    }
    p_queue_comm->size = size;
    p_queue_comm->num = num;
    return (void *)p_queue;
failed:
    if (shm_fd > 0)
    {
        close(shm_fd);
    }

    if (p_queue)
    {
        free(p_queue);
    }

    if (sem_can_read)
    {
        sem_unlink(SEM_READ_NAME(name));
    }

    if (sem_can_write)
    {
        sem_unlink(SEM_WRITE_NAME(name));
    }

    if (shm_ptr)
    {
        munmap(shm_ptr, total_size);
        shm_unlink(name);
    }
    return NULL;
}

void *process_queue_open(char *name, int size, int num, int timeout)
{
    char *shm_ptr = NULL;
    sem_t *sem_can_read = NULL;
    sem_t *sem_can_write = NULL;
    queue_t *p_queue = NULL;
    queue_comm_t *p_queue_comm = NULL;
    int shm_fd = -1, total_size = 0;
    int elaspe_ms = 0, queue_inited = 0;

    p_queue = malloc(sizeof(queue_t));
    if (p_queue == NULL)
    {
        goto failed;
    }

    shm_fd = shm_open(name, O_RDWR, 0666);
    if (shm_fd == -1)
    {
        perror("shm_open");
        goto failed;
    }
    total_size = (size + ITEM_HDR) * num + sizeof(queue_comm_t);
    if (ftruncate(shm_fd, total_size) == -1)
    {
        perror("ftruncate");
        goto failed;
    }
    // 将共享内存映射到进程地址空间
    shm_ptr = mmap(0, total_size, PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
    if (shm_ptr == MAP_FAILED)
    {
        perror("mmap");
        goto failed;
    }
    close(shm_fd);

    p_queue_comm = (queue_comm_t *)shm_ptr;
    if (p_queue_comm->init_flag != INIT_FLAG)
    {
        while (elaspe_ms < timeout)
        {
            delayms(20);
            elaspe_ms += 20;
            if (p_queue_comm->init_flag == INIT_FLAG)
            {
                queue_inited = 1;
            }
        }
    }
    else
    {
        queue_inited = 1;
    }

    if (queue_inited == 0)
    {
        goto failed;
    }

    // 打开命名信号量
    sem_can_read = sem_open(SEM_READ_NAME(name), O_CREAT, 0644, 0);
    if (sem_can_read == SEM_FAILED)
    {
        perror("sem_open1");
        goto failed;
    }

    sem_can_write = sem_open(SEM_WRITE_NAME(name), O_CREAT, 0644, num);
    if (sem_can_write == SEM_FAILED)
    {
        perror("sem_open2");
        goto failed;
    }

    p_queue_comm = (queue_comm_t *)shm_ptr;
    p_queue->comm = p_queue_comm;
    p_queue->data = shm_ptr + sizeof(queue_comm_t);
    p_queue->sem_can_read = sem_can_read;
    p_queue->sem_can_write = sem_can_write;
    return (void *)p_queue;

failed:
    if (shm_fd > 0)
    {
        close(shm_fd);
    }

    if (p_queue)
    {
        free(p_queue);
    }

    if (sem_can_read)
    {
        sem_unlink(SEM_READ_NAME(name));
    }

    if (sem_can_write)
    {
        sem_unlink(SEM_WRITE_NAME(name));
    }

    if (shm_ptr)
    {
        munmap(shm_ptr, total_size);
        shm_unlink(name);
    }
    return NULL;
}

int process_queue_fixup(void *q)
{
    int val1, val2;
    queue_t *queue = (queue_t *)q;
    sem_t *sem_can_read = queue->sem_can_read;
    sem_t *sem_can_write = queue->sem_can_write;
    if (sem_getvalue(sem_can_write, &val1) == 0 && sem_getvalue(sem_can_read, &val2) == 0)
    {
        printf("sem write %d,read %d\n", val1, val2);
    }
    return 0;
}

int process_queue_send(void *q, char *buffer, int size)
{
    char *item;
    queue_t *queue = (queue_t *)q;
    sem_t *sem_can_read = queue->sem_can_read;
    sem_t *sem_can_write = queue->sem_can_write;
    if (queue->comm->init_flag != INIT_FLAG)
    {
        return -1;
    }
    // 队列中有资源可写
    if (sem_wait(sem_can_write) == -1)
    {
        perror("sem_wait");
        return -1;
    }
    // 写用户数据到队列
    item = queue->data + queue->comm->pos_w * (queue->comm->size + ITEM_HDR);
    memset(item, 0x00, queue->comm->size + ITEM_HDR);
    item[0] = (size >> 8) & 0xff;
    item[1] = (size) & 0xff;
    size = (size > (queue->comm->size)) ? queue->comm->size : size;
    memcpy(item + ITEM_HDR, buffer, size);

    // 更新队列可读资源
    queue->comm->pos_w = (queue->comm->pos_w + 1) % queue->comm->num;
    if (sem_post(sem_can_read) == -1)
    {
        perror("sem_post");
        return -1;
    }
    return size;
}

int process_queue_recv(void *q, char *buff, int size)
{
    char *item;
    int item_size;
    queue_t *queue = (queue_t *)q;
    sem_t *sem_can_read = queue->sem_can_read;
    sem_t *sem_can_write = queue->sem_can_write;

    if (queue->comm->init_flag != INIT_FLAG)
    {
        return -1;
    }

    // 队列中有资源可读
    if (sem_wait(sem_can_read) == -1)
    {
        perror("sem_wait");
        return -1;
    }
    // 读队列数据到用户缓存
    item = queue->data + queue->comm->pos_r * (queue->comm->size + ITEM_HDR);
    item_size = (item[0] << 8) + item[1];
    size = (size > item_size) ? item_size : size;
    memcpy(buff, item + ITEM_HDR, size);
    queue->comm->pos_r = (queue->comm->pos_r + 1) % queue->comm->num;
    // 更新队列可写资源
    if (sem_post(sem_can_write) == -1)
    {
        perror("sem_post");
        return -1;
    }

    return size;
}

int process_queue_destory(void *q)
{
    char *ptr;
    int total_size;
    queue_t *queue = (queue_t *)q;
    sem_t *sem_can_read = queue->sem_can_read;
    sem_t *sem_can_write = queue->sem_can_write;

    if (queue->comm->init_flag != INIT_FLAG)
    {
        return 0;
    }

    total_size = (queue->comm->size + ITEM_HDR) * queue->comm->num + sizeof(queue_comm_t);
    if (munmap(queue->data - sizeof(queue_t), total_size) == -1)
    {
        // 处理错误，例如退出程序
        perror("munmap");
    }

    shm_unlink(queue->name);

    // 关闭信号量
    if (sem_close(sem_can_write) == -1)
    {
        perror("sem_close");
    }
    sem_unlink(SEM_WRITE_NAME(queue->name));

    // 关闭信号量
    if (sem_close(sem_can_read) == -1)
    {
        perror("sem_close");
    }
    sem_unlink(SEM_READ_NAME(queue->name));

    // 释放队列结构体
    free(queue);
    return 0;
}

int process_queue_close(queue_t *queue)
{
    char *ptr;
    int total_size;
    sem_t *sem_can_read = queue->sem_can_read;
    sem_t *sem_can_write = queue->sem_can_write;

    if (queue->comm->init_flag != INIT_FLAG)
    {
        return 0;
    }

    total_size = (queue->comm->size + ITEM_HDR) * queue->comm->num + sizeof(queue_t);
    if (munmap(queue->data - sizeof(queue_t), total_size) == -1)
    {
        // 处理错误，例如退出程序
        perror("munmap");
    }

    // 关闭信号量
    if (sem_close(sem_can_write) == -1)
    {
        perror("sem_close");
    }
    // 关闭信号量
    if (sem_close(sem_can_read) == -1)
    {
        perror("sem_close");
    }

    // 释放队列结构体
    free(queue);
    return 0;
}